package com.kai.mqtt.handler;

import com.kai.mqtt.constant.AttributeKeyConstant;
import com.kai.mqtt.message.WillMessage;
import com.kai.mqtt.message.service.*;
import com.kai.mqtt.protocol.AbstractProtocol;
import com.kai.mqtt.protocol.Publish;
import com.kai.mqtt.session.Session;
import com.kai.mqtt.session.service.ISessionService;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateEvent;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Objects;

/**
 * @author weijk
 * @since 2024/6/2
 */
@Slf4j
@Scope("prototype")
@Component
public class BrokerHandler extends SimpleChannelInboundHandler<MqttMessage> {

    @Autowired
    private ApplicationContext applicationContext;

    @Resource
    private ISessionService sessionService;

    @Resource
    private IWillMessageService willMessageService;

    @Resource
    private ISubscribeMessageService subscribeMessageService;

    @Resource
    private IMessageIdService messageIdService;

    @Resource
    private IDupPublishMessageService dupPublishMessageService;

    @Resource
    private IDupPubRelMessageService dupPubRelMessageService;

    @Resource
    private Publish publish;

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        String clientId = ctx.channel().attr(AttributeKeyConstant.CLIENT_ID).get();
        log.debug("通道断开连接，clientId: {}", clientId);
        if (StringUtils.hasText(clientId)) {
            synchronized (clientId.intern()) {
                clientId = ctx.channel().attr(AttributeKeyConstant.CLIENT_ID).get();
                if (!StringUtils.hasText(clientId)) {
                    return;
                }

                Session session = sessionService.get(clientId);
                if (Objects.nonNull(session) && session.getIsCleanSession()) {
                    willMessageService.remove(clientId);
                    subscribeMessageService.remove(clientId);
                    messageIdService.remove(clientId);
                    dupPublishMessageService.remove(clientId);
                    dupPubRelMessageService.remove(clientId);
                    sessionService.remove(clientId);
                }
            }
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
        Class<? extends AbstractProtocol> clazz = AbstractProtocol.MessageEnum.getClassByMessageType(msg.fixedHeader().messageType());
        if (Objects.isNull(clazz)) {
            log.info("未匹配到消息处理协议，messageType: {}", msg.fixedHeader().messageType().name());
            return;
        }
        AbstractProtocol abstractProtocol = applicationContext.getBean(clazz);
        abstractProtocol.process(ctx.channel(), msg);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        String clientId = ctx.channel().attr(AttributeKeyConstant.CLIENT_ID).get();
        log.warn("通道事件触发，clientId: {}", clientId);
        if (evt instanceof IdleStateEvent) {
            this.sendWillMessage(ctx.channel());
        }
        ctx.channel().close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        String clientId = ctx.channel().attr(AttributeKeyConstant.CLIENT_ID).get();
        log.error("通道异常，clientId: {}, message: {}", clientId, cause.getMessage(), cause);
        this.sendWillMessage(ctx.channel());
        ctx.channel().close();
    }

    private void sendWillMessage(Channel channel) {
        String clientId = channel.attr(AttributeKeyConstant.CLIENT_ID).get();
        WillMessage willMessage = willMessageService.get(clientId);
        if (Objects.nonNull(willMessage)) {
            MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.valueOf(willMessage.getQos()), willMessage.getIsRetain(), 0),
                    new MqttPublishVariableHeader(willMessage.getTopic(), 0), Unpooled.buffer().writeBytes(willMessage.getMessageBytes()));
            publish.process(channel, publishMessage);
        }
    }
}
